package rdd01.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Anli_Test {
  def main(args: Array[String]): Unit = {
    val sparkConf=new SparkConf().setMaster("local[*]").setAppName("combineByKey")
    val sc=new SparkContext(sparkConf)
    //统计每个省份每个广告被点击的数量的top3
    //数据规则:时间戳 省份 城市 用户 广告
    /**
     * 问题解决思路：
     *   把缺少的东西补上，把多余的东西删除过滤
     *   数据聚合一般需要补充相应的东西
     *步骤：
     *   1、获取原始数据：时间戳 省份 城市 用户 广告
     *   2、将原始数据进行结构的转换，方便统计（想要什么就转换什么）=>((省份，广告)，1)
     *   3、将转换结构后的数据进行分组聚合=>((省份，广告)，1)=>((省份，广告)，sum)
     *   4、将聚合的结果进行结构的转换=>((省份，广告)，sum)=>(省份，(广告，sum))
     *   5、将转换结构后的数据根据省份进行分组=>(省份，【（广告A，sumA）,（广告B，sumB）】)
     *   6、将数据进行内排序，取前3
     *   7、采集数据进行控制台打印
     *   */
    val dataRDD=sc.textFile("data/agent.log")
    println(dataRDD.toDebugString)
    println(">>>>>>>>>>>>>>>>>>>>>>>>>>")
    //((省份，广告),1)
//    2、将原始数据进行结构的转换，方便统计（想要什么就转换什么）=>((省份，广告)，1)
    val mapRDD: RDD[((String, String), Int)] = dataRDD.map(
      line => {
        val datas: Array[String] = line.split(" ")
        ((datas(1), datas(4)), 1)
      }
    )
    println(mapRDD.toDebugString)
    println(">>>>>>>>>>>>>>>>>>>>>>>>>>")
   //分组聚合   ((省份，广告)，1)=>((省份，广告)，sum)
    //(省份1，广告1)+  (省份1，广告1)
//    3、将转换结构后的数据进行分组聚合=>((省份，广告)，1)=>((省份，广告)，sum)
   val reduceRDD: RDD[((String, String), Int)] = mapRDD.reduceByKey(_ + _)
    println(reduceRDD.toDebugString)
    println(">>>>>>>>>>>>>>>>>>>>>>>>>>")
//   val reduceRDD1: RDD[((String, String), Int)] = mapRDD.reduceByKey((x,y)=>{ x+y})
   //((省份，广告)，sum)=>(省份，(广告，sum))
//    4、将聚合的结果进行结构的转换=>((省份，广告)，sum)=>(省份，(广告，sum))
   val newMapRDD: RDD[(String, (String, Int))] = reduceRDD.map {
     //模式匹配规则进行转换,转换成想要的数据结构
     case ((prv, ad), sum) => {
       (prv, (ad, sum))
     }
   }
    println(newMapRDD.toDebugString)
    println(">>>>>>>>>>>>>>>>>>>>>>>>>>")
//    5、将转换结构后的数据根据省份进行分组=>(省份，【（广告A，sumA）,（广告B，sumB）】)
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = newMapRDD.groupByKey()
//    6、将数据进行内排序，取前3
val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
  iter => {
    iter.toList.sortBy(_._2).take(3)
  }
)


    println(resultRDD.toDebugString)
    println(">>>>>>>>>>>>>>>>>>>>>>>>>>")
    resultRDD.collect().foreach(println)
    sc.stop()

  }

}
